package transpond

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"net/http"
	"runtime"
	"sync"
	"time"

	"github.com/cloudwego/hertz/pkg/common/hlog"
)

// -----------------------------------------------------------------------------
// HTTP 发送器
// -----------------------------------------------------------------------------
type Sender struct {
	queue      *Queue
	client     *http.Client
	workers    int
	retryMax   int
	retryDelay time.Duration
	wg         sync.WaitGroup
}

func NewSender(q *Queue) *Sender {
	return &Sender{
		queue: q,
		client: &http.Client{
			Transport: &http.Transport{
				MaxIdleConns:        100,
				MaxIdleConnsPerHost: 100,
			},
			Timeout: 10 * time.Second,
		},
		workers:    runtime.NumCPU() * 4,
		retryMax:   5,
		retryDelay: time.Second,
	}
}

// 启动发送器
func (s *Sender) Start(ctx context.Context) {
	sem := make(chan struct{}, s.workers)
	tick := time.NewTicker(1000 * time.Millisecond)
	defer tick.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-tick.C:
			reqs := s.queue.Dequeue(s.workers * 2)
			if len(reqs) == 0 {
				continue
			}
			hlog.Info("dequeue request: %v", reqs)
			for _, r := range reqs {
				sem <- struct{}{}
				s.wg.Add(1)
				go func(req *PendingRequest) {
					defer func() {
						<-sem
						s.wg.Done()
					}()
					req.Status = "processing"
					req.ProcessingStartTime = time.Now()
					s.sendWithRetry(req)
				}(r)
			}
		}
	}
}

func (s *Sender) Stop() {
	s.wg.Wait()
}

// 发送请求，支持重试
func (s *Sender) sendWithRetry(req *PendingRequest) {
	req.Status = "processing"
	req.ProcessingStartTime = time.Now()
	err := s.send(req)
	if err == nil {
		hlog.Info("send request success: %s", req.ID)
		s.queue.Delete(req.ID) // 成功：队列中已删除
		return
	}
	req.Retry++
	if req.Retry > s.retryMax {
		hlog.Info("dead letter: %s err=%v", req.ID, err)
		return
	}
	// 指数退避
	delay := s.retryDelay * time.Duration(1<<(req.Retry-1))
	req.NextTry = time.Now().Add(delay)
	hlog.Info("send request failed: %s err=%v retry=%d nextTry=%s", req.ID, err, req.Retry, req.NextTry)
	if err := s.queue.Update(req); err != nil {
		hlog.Info("failed to reschedule: %v", err)
	}
}

func (s *Sender) send(req *PendingRequest) error {
	httpReq, _ := http.NewRequest(req.Method, req.URL, bytes.NewReader(req.Body))
	httpReq.Header = req.Headers

	resp, err := s.client.Do(httpReq)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	if resp.StatusCode >= 500 {
		body, _ := io.ReadAll(resp.Body)
		return fmt.Errorf("status=%d body=%q", resp.StatusCode, body)
	}
	_, _ = io.Copy(io.Discard, resp.Body) // 必须读完
	return nil
}
